package com.example.demo.service;

import com.alibaba.fastjson.JSON;
import com.example.demo.model.Order;
import com.example.demo.model.PaymentRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.time.format.DateTimeFormatter;
import java.util.Optional;

/**
 * 消费交易数据
 */
@Component
public class OrderConsumer {
    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 监听Kafka的dewuOrder主题，订单消息
     * @param record
     */
    @KafkaListener(topics = "dewuOrder")
    public void listen(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record);
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            Order order = JSON.parseObject(message.toString(), Order.class);
            String date = order.getGmtCreated().format(DateTimeFormatter.ofPattern("yyyyMMdd"));
            RLock lock = redissonClient.getLock("orderLock-" + date);
            try {
                lock.lock();
                Double number = (Double) redisTemplate.opsForValue().get("order-" + date);
                if (number == null) {
                    number = 0.0;
                }
                number = number + order.getTotalPrice();
                redisTemplate.opsForValue().set("order-" + date, number);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }

        }
    }

    /**
     * 监听Kafka的dewuPayment主题，支付消息
     * @param record
     */
    @KafkaListener(topics = "dewuPayment")
    public void listen2(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record);
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            PaymentRecord paymentRecord = JSON.parseObject(message.toString(), PaymentRecord.class);
            String date = paymentRecord.getGmtCreated().format(DateTimeFormatter.ofPattern("yyyyMMdd"));
            RLock lock = redissonClient.getLock("paymentLock-" + date);
            try {
                lock.lock();
                Double number = (Double) redisTemplate.opsForValue().get("payment-" + date);
                if (number == null) {
                    number = 0.0;
                }
                number = number + paymentRecord.getAmount();
                redisTemplate.opsForValue().set("payment-" + date, number);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }
}
